// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/testing/gtest_utils.h>
#include <pollux/common/file/file.h>
#include <pollux/connectors/hive/file_handle.h>
#include <pollux/connectors/hive/storage_adapters/s3fs/S3FileSystem.h>
#include <pollux/connectors/hive/storage_adapters/s3fs/S3Util.h>
#include <pollux/connectors/hive/storage_adapters/s3fs/tests/MinioServer.h>
#include <pollux/testing/exec/util/hive_connector_test_base.h>
#include <pollux/common/file/temp_file_path.h>

#include <gtest/gtest.h>

using namespace kumo::pollux;

constexpr int kOneMB = 1 << 20;

static constexpr std::string_view kDummyPath = "s3://dummy/foo.txt";

class S3Test : public testing::Test {
 protected:
  void SetUp() override {
    minioServer_ = std::make_unique<MinioServer>();
    minioServer_->start();
    ioExecutor_ = std::make_unique<melon::IOThreadPoolExecutor>(3);
  }

  void TearDown() override {
    minioServer_->stop();
  }

  void addBucket(const char* bucket) {
    minioServer_->addBucket(bucket);
  }

  std::string localPath(const char* directory) {
    return minioServer_->path() + "/" + directory;
  }

  void writeData(WriteFile* writeFile) {
    writeFile->append("aaaaa");
    writeFile->append("bbbbb");
    writeFile->append(std::string(kOneMB, 'c'));
    writeFile->append("ddddd");
    ASSERT_EQ(writeFile->size(), 15 + kOneMB);
  }

  void readData(ReadFile* readFile) {
    ASSERT_EQ(readFile->size(), 15 + kOneMB);
    char buffer1[5];
    ASSERT_EQ(readFile->pread(10 + kOneMB, 5, &buffer1), "ddddd");
    char buffer2[10];
    ASSERT_EQ(readFile->pread(0, 10, &buffer2), "aaaaabbbbb");
    char buffer3[kOneMB];
    ASSERT_EQ(readFile->pread(10, kOneMB, &buffer3), std::string(kOneMB, 'c'));
    ASSERT_EQ(readFile->size(), 15 + kOneMB);
    char buffer4[10];
    const std::string_view arf = readFile->pread(5, 10, &buffer4);
    const std::string zarf = readFile->pread(kOneMB, 15);
    auto buf = std::make_unique<char[]>(8);
    const std::string_view warf = readFile->pread(4, 8, buf.get());
    const std::string_view warfFromBuf(buf.get(), 8);
    ASSERT_EQ(arf, "bbbbbccccc");
    ASSERT_EQ(zarf, "ccccccccccddddd");
    ASSERT_EQ(warf, "abbbbbcc");
    ASSERT_EQ(warfFromBuf, "abbbbbcc");
    char head[12];
    char middle[4];
    char tail[7];
    std::vector<melon::Range<char*>> buffers = {
        melon::Range<char*>(head, sizeof(head)),
        melon::Range<char*>(nullptr, (char*)(uint64_t)500000),
        melon::Range<char*>(middle, sizeof(middle)),
        melon::Range<char*>(
            nullptr,
            (char*)(uint64_t)(15 + kOneMB - 500000 - sizeof(head) -
                              sizeof(middle) - sizeof(tail))),
        melon::Range<char*>(tail, sizeof(tail))};
    ASSERT_EQ(15 + kOneMB, readFile->preadv(0, buffers));
    ASSERT_EQ(std::string_view(head, sizeof(head)), "aaaaabbbbbcc");
    ASSERT_EQ(std::string_view(middle, sizeof(middle)), "cccc");
    ASSERT_EQ(std::string_view(tail, sizeof(tail)), "ccddddd");
  }

  std::unique_ptr<MinioServer> minioServer_;
  std::unique_ptr<melon::IOThreadPoolExecutor> ioExecutor_;
};
